package com.fxu.framework.core.redis;

import cn.hutool.core.io.IoUtil;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import net.jpountz.lz4.*;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.SerializationException;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.stereotype.Component;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

/**
 * 配置Redis，使用Json格式化，LZ4压缩、Gzip压缩
 * @author fangxu
 * @version 1.0.0
 */
@Component
public class RedisConfig {
	/**
	 * <p>重写Redis序列化方式，使用Json方式:</p>
	 * <p>当我们的数据存储到Redis的时候，我们的键（key）和值（value）都是通过Spring提供的Serializer序列化到数据库的。
	 * RedisTemplate默认使用的是JdkSerializationRedisSerializer，StringRedisTemplate默认使用的是StringRedisSerializer。</p>
	 * <p>Spring Data JPA为我们提供了下面的Serializer：</p>
	 * <p>GenericToStringSerializer、Jackson2JsonRedisSerializer、JacksonJsonRedisSerializer、
	 * JdkSerializationRedisSerializer、OxmSerializer、StringRedisSerializer。</p>
	 * <p>在此我们将自己配置RedisTemplate并定义Serializer。</p>
	 * @param redisConnectionFactory redis连接工厂对象
	 * @return redis连接模版对象
	 */
	@Bean
	public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
		RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
		redisTemplate.setConnectionFactory(redisConnectionFactory);

		Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = getJackson2JsonRedisSerializer();

		// 设置值（value）的序列化采用Jackson2JsonRedisSerializer。
		redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
//		redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);
//		redisTemplate.setValueSerializer(new GzipSerializer(jackson2JsonRedisSerializer));
		redisTemplate.setHashValueSerializer(new LZ4Serializer(jackson2JsonRedisSerializer));
		// 设置键（key）的序列化采用StringRedisSerializer。
		redisTemplate.setKeySerializer(new StringRedisSerializer());
		redisTemplate.setHashKeySerializer(new StringRedisSerializer());

		redisTemplate.afterPropertiesSet();

		return redisTemplate;
	}

	@SuppressWarnings("deprecation")
	public static Jackson2JsonRedisSerializer<Object> getJackson2JsonRedisSerializer() {
		Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<>(Object.class);
		ObjectMapper om = new ObjectMapper();
		om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
		om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
		om.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); // 忽略不对应字段
		jackson2JsonRedisSerializer.setObjectMapper(om);
		return jackson2JsonRedisSerializer;
	}

	/**
	 * 高效压缩解压
	 */
	public static class LZ4Serializer implements RedisSerializer<Object> {

		private static final int BUFFER_SIZE = 4096;

		private RedisSerializer<Object> innerSerializer;
		private LZ4FastDecompressor decompresser;
		private LZ4Compressor compressor;

		public LZ4Serializer(RedisSerializer<Object> innerSerializer) {
			this.innerSerializer = innerSerializer;
			LZ4Factory factory = LZ4Factory.fastestInstance();
			this.compressor = factory.fastCompressor();
			this.decompresser = factory.fastDecompressor();
		}

		@Override
		public byte[] serialize(Object graph) throws SerializationException {
			if (graph == null) {
				return new byte[0];
			}
			ByteArrayOutputStream byteOutput = null;
			LZ4BlockOutputStream compressedOutput = null;
			try {
				byte[] bytes = innerSerializer.serialize(graph);
				byteOutput = new ByteArrayOutputStream();
				compressedOutput = new LZ4BlockOutputStream(byteOutput, BUFFER_SIZE, compressor);
				compressedOutput.write(bytes);
				compressedOutput.finish();
				return byteOutput.toByteArray();
			} catch (Exception e) {
				throw new SerializationException("LZ4 Serialization Error", e);
			} finally {
				IoUtil.close(compressedOutput);
				IoUtil.close(byteOutput);
			}
		}

		@Override
		public Object deserialize(byte[] bytes) throws SerializationException {

			if (bytes == null || bytes.length == 0) {
				return null;
			}

			ByteArrayOutputStream baos = null;
			LZ4BlockInputStream lzis = null;
			try {
				baos = new ByteArrayOutputStream(BUFFER_SIZE);
				lzis = new LZ4BlockInputStream(new ByteArrayInputStream(bytes), decompresser);
				int count;
				byte[] buffer = new byte[BUFFER_SIZE];
				while ((count = lzis.read(buffer)) != -1) {
					baos.write(buffer, 0, count);
				}
				return innerSerializer.deserialize(baos.toByteArray());
			} catch (Exception e) {
				throw new SerializationException("LZ4 deserizelie error", e);
			} finally {
				IoUtil.close(lzis);
				IoUtil.close(baos);
			}
		}
	}

	static class GzipSerializer implements RedisSerializer<Object> {

		public static final int BUFFER_SIZE = 4096;
		// 这里组合方式，使用到了一个序列化器
		private RedisSerializer<Object> innerSerializer;

		public GzipSerializer(RedisSerializer<Object> innerSerializer) {
			this.innerSerializer = innerSerializer;
		}

		@Override
		public byte[] serialize(Object graph) throws SerializationException {
			if (graph == null) {
				return new byte[0];
			}
			ByteArrayOutputStream bos = null;
			GZIPOutputStream gzip = null;
			try {
				// 先序列化
				byte[] bytes = innerSerializer.serialize(graph);
				bos = new ByteArrayOutputStream();
				gzip = new GZIPOutputStream(bos);
				// 在压缩
				gzip.write(bytes);
				gzip.finish();
				return bos.toByteArray();
			} catch (Exception e) {
				throw new SerializationException("Gzip Serialization Error", e);
			} finally {
				IoUtil.close(bos);
				IoUtil.close(gzip);
			}
		}

		@Override
		public Object deserialize(byte[] bytes) throws SerializationException {

			if (bytes == null || bytes.length == 0) {
				return null;
			}

			ByteArrayOutputStream bos = null;
			ByteArrayInputStream bis = null;
			GZIPInputStream gzip = null;
			try {
				bos = new ByteArrayOutputStream();
				bis = new ByteArrayInputStream(bytes);
				gzip = new GZIPInputStream(bis);
				byte[] buff = new byte[BUFFER_SIZE];
				int n;
				// 先解压
				while ((n = gzip.read(buff, 0, BUFFER_SIZE)) > 0) {
					bos.write(buff, 0, n);
				}
				// 再反序列化
				return innerSerializer.deserialize(bos.toByteArray());
			} catch (Exception e) {
				throw new SerializationException("Gzip deserizelie error", e);
			} finally {
				IoUtil.close(bos);
				IoUtil.close(bis);
				IoUtil.close(gzip);
			}
		}
	}
}
